package zy.learn.demo.structuredstreaming.watermark

import java.sql.Timestamp

import org.apache.spark.SparkConf
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.streaming.OutputMode

/**
 * 没有窗口，使用水印，没有聚合，数据可以直接 select出来
 * 没有窗口，且不以水印的时间字段使用水印没有效果，事件时间小于等于当前水印的数据不会被Select出来参加聚合运算
 */
object WatermarkNoWindow {
  def main(args: Array[String]): Unit = {
    val sparkConf = new SparkConf().set("spark.sql.shuffle.partitions", "3")
    val spark = SparkSession.builder()
      .master("local[2]")
      .config(sparkConf)
      .appName("Watermark Tumbling Window start")
      .getOrCreate()

    import spark.implicits._

    val lines = spark.readStream
      .format("socket") // 设置数据源
      .option("host", "co7-203")
      .option("port", 9999)
      .load
    /* 输入的数据
    * 2020-10-14 10:55:00,dog     # wartermark: 10:55:00 - 2 min = 10:53:00
    * 2020-10-14 11:00:00,dog     # wartermark: 11:00:00 - 2 min = 10:58:00
    * 2020-10-14 10:55:00,dog     # wartermark: 10:55:00 - 2 min = 10:53:00 < 10:58:00 => 10:58:00
    * 2020-10-14 11:05:00,dog     # wartermark: 11:00:00 - 2 min = 11:03:00
    * 2020-10-14 10:55:00,dog
    * 2020-10-14 11:20:00,dog     # wartermark: 11:00:00 - 2 min = 11:18:00
    * 2020-10-14 10:55:00,dog
    *
    * 不做聚合，直接输出的结果
+-------------------+----+
|ts                 |word|
+-------------------+----+
|2020-10-14 10:55:00|dog |  Batch: 0
+-------------------+----+
|2020-10-14 11:00:00|dog |  Batch: 1
+-------------------+----+
|2020-10-14 10:55:00|dog |  Batch: 2
+-------------------+----+
|2020-10-14 11:05:00|dog |  Batch: 3
+-------------------+----+
|2020-10-14 10:55:00|dog |  Batch: 4
+-------------------+----+
|2020-10-14 11:20:00|dog |  Batch: 5
+-------------------+----+
|2020-10-14 10:55:00|dog |  Batch: 6
+-------------------+----+
*
* 输入的数据
    * 2020-10-14 10:55:00,dog     # wartermark: 10:55:00 - 2 min = 10:53:00
    * 2020-10-14 11:00:00,dog     # wartermark: 11:00:00 - 2 min = 10:58:00
    * 2020-10-14 10:55:00,dog     # wartermark: 10:55:00 - 2 min = 10:53:00 < 10:58:00 => 10:58:00
    * 2020-10-14 11:05:00,dog     # wartermark: 11:00:00 - 2 min = 11:03:00
    * 2020-10-14 11:04:00,dog     # wartermark: 11:03:00
    * 2020-10-14 11:05:00,dog     # wartermark: 11:00:00 - 2 min = 11:03:00
    * 2020-10-14 11:04:00,dog     # wartermark: 11:03:00
    * 2020-10-14 11:03:00,dog     # wartermark: 11:03:00
* select ts, word, count(*) from table group by ts, word
* 输出结果
Batch: 0
+-------------------+----+--------+
|ts                 |word|count(1)|
+-------------------+----+--------+
|2020-10-14 10:55:00|dog |1       |
+-------------------+----+--------+
Batch: 1
+-------------------+----+--------+
|ts                 |word|count(1)|
+-------------------+----+--------+
|2020-10-14 11:00:00|dog |1       |
+-------------------+----+--------+
Batch: 2
+---+----+--------+
|ts |word|count(1)|
+---+----+--------+
+---+----+--------+
Batch: 3
+-------------------+----+--------+
|ts                 |word|count(1)|
+-------------------+----+--------+
|2020-10-14 11:05:00|dog |1       |
+-------------------+----+--------+
Batch: 4
+-------------------+----+--------+
|ts                 |word|count(1)|
+-------------------+----+--------+
|2020-10-14 11:04:00|dog |1       |
+-------------------+----+--------+
Batch: 5
+-------------------+----+--------+
|ts                 |word|count(1)|
+-------------------+----+--------+
|2020-10-14 11:05:00|dog |2       |
+-------------------+----+--------+
Batch: 6
+-------------------+----+--------+
|ts                 |word|count(1)|
+-------------------+----+--------+
|2020-10-14 11:04:00|dog |2       |
+-------------------+----+--------+
Batch: 7
+---+----+--------+
|ts |word|count(1)|
+---+----+--------+
+---+----+--------+
    * */
    val wordsDF = lines.as[String].map(line => {
      val split = line.split(",")
      (Timestamp.valueOf(split(0)), split(1))
    }).toDF("ts", "word")

    import org.apache.spark.sql.functions._

    wordsDF
      .withWatermark("ts", "2 minutes")
      .createOrReplaceTempView("table")
    /*val wordCounts = spark.sql(
      """
        |select ts, word from table
        |""".stripMargin)*/
    val wordCounts = spark.sql(
      """
        |select ts, word, count(*) from table group by ts, word
        |""".stripMargin)

    val query = wordCounts.writeStream
      .format("console")
      .outputMode(OutputMode.Update())
      .option("truncate", "false")
      .start()

    query.awaitTermination()
  }
}
